package com.stream

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

object FlinkSocket {
  def main(args: Array[String]): Unit = {
    // 1. 获取Flink执行环境,批处理(batch jon)用ExecutionEnvironment,流处理(stream job) 用StreamExecutionEnvironment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 2. 创建数据源
    // 3. 获取数据源
//    val port = 9999;
//    val host = "192.168.86.101"
    val port = 9000;
    val host = "localhost"
    val source = env.socketTextStream(host,port)
    // 4.  解析socket数据源,进行单词分开统计
     val dataStream = source
          .flatMap(_.split(" "))   // 获取数据并进行切分，生成一个个单词
          .map((_,1))                      // 将一个个单词生成一个个对偶元组
          .keyBy(0)                //启动0代表元组的下标,"0"下标代表key: 单词
          .timeWindow(Time.seconds(2),Time.seconds(2)) //指定计算数据的窗口大小和滑动窗口大小
          .sum(1)                // 其中1代表元组中的下标,"1"下标代表: 单词出现的次数
        //输出到目的端
        dataStream.print()
    env.execute("FlinkSocket ")
  }

}
